package zsw.rocketmqs.Demo;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import web.dao.AccountMapper;
import web.dao.TmessageTransactionMapper;
import zsw.model.Account;

@RestController
public class ProducerDemo {
	private static final Logger log = Logger.getLogger(ProducerDemo.class);

	@Autowired
	private DefaultMQProducer defaultProducer;

	@Autowired
	private TransactionMQProducer transactionProducer;

	private static ExecutorService executorService;
	
	@Autowired
	private AccountMapper mapper;
	
	@Autowired
	private TransactionListener orderTransactionListenerImpl;
	
	   @Autowired
private TmessageTransactionMapper tmapper;

	private volatile int i = 0;

	@RequestMapping(value = "/sendMsg", method = RequestMethod.GET)
	public String sendMsg() {

		/**
		 * 发送的是同步消息，上一条消息没有应答，将不会发送下一条消息.消息不会丢失
		 */

		for (int ii = 0; ii < 100000000; ii++) {
			Message msg = new Message("TopicTest1", "TagA", "OrderID00" + i, ("Hello zebra mq" + i).getBytes());

			try {
				SendResult sendResult = defaultProducer.send(msg);
				if (sendResult != null) {
					System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic()
							+ " msgId is: " + sendResult.getMsgId());
				}
				// 同步发送消息，只要不抛异常就是成功
			} catch (Exception e) {
				// 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
				System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
				e.printStackTrace();
			}
			i++;
		}
		return "xds";
	}

	@RequestMapping(value = "/sendMsgAsy", method = RequestMethod.GET)
	public String sendMsg2() throws InterruptedException {
		/**
		 * 异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。然后回调，接收 服务器的响应。消息不会丢失
		 */
		i = 0;
		createThreadPool();
		defaultProducer.setCallbackExecutor(executorService);
		final CountDownLatch ds = new CountDownLatch(100000);
		for (int ii = 1; ii <= 1000000; ii++) {

			/*
			 * if(ii%10000==0) { Thread.sleep(1000L); }
			 */
			Message msg = new Message("TopicTest1", // topic
					"TagA", // tag
					"OrderID00" + i, // key
					("Hello zebra mq" + i).getBytes());// body

			try {

				defaultProducer.send(msg, new SendCallback() {

					@Override
					public void onSuccess(SendResult sendResult) {
						System.out.println(sendResult);
						System.out.println("成功发送消息" + (i++) + sendResult); // TODO 发送成功处理
						ds.countDown();
					}

					@Override
					public void onException(Throwable e) {
						// 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处 理
						System.out.println(e); // TODO
						// 发送失败处理
					}
				}, 60 * 1000L);
			} catch (Exception e) {
				e.printStackTrace();
			}

		}
		ds.await();
		System.out.println("******************完成");
		return "fcrgt";
	}

	@RequestMapping(value = "/sendMsgOneWay", method = RequestMethod.GET)
	public String sendMsg3() {
		/**
		 * 单向发送，不需要等待服务器的响应，可能存在消息的丢失
		 */
		for (int ii = 0; ii < 100000; ii++) {
			Message msg = new Message("TopicTest1", // topic
					"TagA", // tag
					"OrderID00" + i, // key
					("Hello zebra mq" + i).getBytes());// body

			try {
				defaultProducer.sendOneway(msg);
			} catch (Exception e) {
				// TODOAuto-generated catch block
				e.printStackTrace();
			}
			i++;
		}
		return "cdcdafafc" + i;
	}
	
	@Transactional
	@RequestMapping(value = "/sendTransactionMsg", method = RequestMethod.GET)
	public String sendTransactionMsg() throws Exception  {
		SendResult sendResult = null;
			// 构造消息
			Account a=new Account();
			a.setMoneyNum(500);
			a.setUserName("A");
			mapper.updateAcount(a);
			
			Message msg = new Message("TopicTest1", // topic
					"TagA", // tag
					"OrderID001", // key
					("500").getBytes());// body

			// 发送事务消息，LocalTransactionExecute的executeLocalTransactionBranch方法中执行本地逻辑
			msg.setTransactionId(UUID.randomUUID().toString().replaceAll("-", ""));
			transactionProducer.setTransactionListener(orderTransactionListenerImpl);
			sendResult=	transactionProducer.sendMessageInTransaction(msg, null);
			Thread.sleep(60*1000L);
			int s=1/0;
		    return sendResult.toString();
	}

	@RequestMapping(value = "/sendMsgOrder", method = RequestMethod.GET)
	public void sendMsgOrder() {

		String[] msgs = new String[] { "1订单消息", "1支付消息", "2订单消息", "1订单完成消息", "3订单消息", "2支付消息", "3支付消息", "2订单完成消息",
				"4订单消息", "3订单完成消息", "4支付消息", "4订单完成消息", "5订单消息", "6订单消息", "6支付消息", "5支付消息", "5订单完成消息", "6订单完成消息"

		};

		for (int ii = 0; ii < msgs.length; ii++) {
			Integer orderId = new Integer(msgs[ii].substring(0, 1));
			Message msg = new Message("TopicTest1", // topic
					"TagA", // tag
					"OrderID00" + i, // key
					(msgs[ii]).getBytes());// body

			try {
				defaultProducer.send(msg, new MessageQueueSelector() {
					@Override
					public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
						int index = ((Integer) arg) % mqs.size();
						System.out.println("MessageQueue" + arg + ":" + index);
						return mqs.get(index);
					}
				}, orderId);// i==arg
				i++;
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

	}

	@RequestMapping(value = "/sendMsgDelay", method = RequestMethod.GET)
	public void sendMsgDelay() {

		/**
		 * 延迟消息
		 * 
		 * ： “1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h
		 * 2h”；若要发送定时消息，在应用层初始化Message消息对象之后，调用Message.setDelayTimeLevel(int
		 * level)方法来设置延迟级别， 按照序列取相应的延迟级别，例如level=2，则延迟为5s；
		 * 对于延迟级别的设置只有在该消息为非事务性消息（sysflag的第3/4字节为0）或者为提交事务消息
		 * （sysflag的第3字节为0，第4字节为1）时才会生效，默认情况下消息是非事务性消息。 ---------------------
		 */

		Message msg = new Message("TopicTest1", // topic
				"TagA", // tag
				"OrderID00" + i, // key
				("Hello zebra mq" + i).getBytes());// body
		msg.setDelayTimeLevel(2); // 延迟5S

		try {
			defaultProducer.send(msg, new MessageQueueSelector() {
				@Override
				public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
					System.out.println("MessageQueue" + arg);
					int index = ((Integer) arg) % mqs.size();
					return mqs.get(index);
				}
			}, i);// i==arg
			i++;
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	@RequestMapping(value = "/sendMsgFilter", method = RequestMethod.GET)
	public void sendMsgOntime() {
		
		for (int i = 0; i < 100; i++) {
			Message msg = new Message("TopicTest1", // topic
					"TagA", // tag
					"OrderID00" + i, // key
					("Hello zebra mq:" + i).getBytes());// body
			
		//	int d=i%2;
			
		//	msg.putUserProperty("a", d+"");
			msg.putUserProperty("ids", i+"");
			try {
				defaultProducer.send(msg);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
		
	}
	
	

	private synchronized static void createThreadPool() {
		if (executorService != null)
			return;

		BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(150);
		ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 0L, TimeUnit.SECONDS, queue);
		threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {

			public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
				try {
					executor.getQueue().put(r);
					log.info("此时的状态是，让生产者阻塞的去提交任务！");
					System.out.println("此时的状态是，让生产者阻塞的去提交任务！");

				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		});

		executorService = threadPool;
	}

}
